package com.talent.study;

import cn.hutool.json.JSONObject;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.web.client.RestTemplate;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;

/**
 * @ClassName：Test
 * @Author: hszhou
 * @Date: 2025/1/13 18:57
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class Test {


    private   static FluxSink<String> outSink;


   private static Consumer<String> producer;

    private static void test01(){
        Flux.empty();

        Mono.empty();

        Flux.error(new RuntimeException("error"));

        Mono.error(new RuntimeException("error"));

        Flux.never();
        Mono.never();

        Flux.just("foo","bar","foobar");
        Mono.just("foo");

        Flux.range(1,10);

        Flux.fromArray(new String[]{"a","b"});

        List<String> ls = new ArrayList<>();
        Flux.fromStream(ls.stream());

        Mono.fromFuture(CompletableFuture.completedFuture("foo"));

        Mono.fromRunnable(()-> System.out.println("goo"));

        Flux.interval(Duration.ofSeconds(1));

    }

    private static void test02(){
                Flux.generate(synchronousSink -> {

            synchronousSink.next("hello");
        }).subscribe(s->{
            System.out.println(Thread.currentThread().getName()+"==="+s);
        });


        Flux.generate(()->0,(state,sink)->{
            sink.next(state+1);
            if(state==10){
                sink.complete();
            }
            return state+1;
        }).subscribe(
                s-> System.out.println("消费者获取的值"+s),
                err-> System.out.println("遇到错误"+err.toString()),
                ()-> System.out.println("complete")
        );
        Flux.generate(()->0,(state,sink)->{
            sink.next(state+1);
            if(state==10){
                sink.complete();
            }
            return state+1;
        },state->{
            System.out.println("获取最后到状态"+state);
        }).subscribe(
                s-> System.out.println("消费者获取的值"+s),
                err-> System.out.println("遇到错误"+err.toString()),
                ()-> System.out.println("complete")
        );
    }

    private static void test03(){

        Flux<String> f = Flux.create(sink -> {
            outSink = sink;
        });
        f.subscribe(e -> System.out.println(e));
        //下发元素
        outSink.next("我来了");
        outSink.complete();
        outSink.next("我来了2");

    }

    private static void test04(){
        Flux.create(sink -> {
            producer = nextData -> sink.next(nextData);
        }).subscribe(e -> System.out.println(e));

        //do something

        //下发元素
        producer.accept("我来了");

        producer.accept("我来了2");
    }


    private static void  test05(){
        Flux.just(1,2,3)
                .map(i->i*2)
                .map(i->i+1);
      //  Flux.just("a","b","c").filter(s->s.equals("a"));
//flatMap
      //  Flux.just("a","b","c").flatMap(s->Flux.just(s.toUpperCase()))
          //      .subscribe(s-> System.out.println(s));
//reduce
        Flux.just(1,2,3).reduce(10,(a,b)->a+b)
                .subscribe(s-> System.out.println(s));

      //  Flux.just("a","b","c").map(s->s.equals("a"));

    }

    private static void  test06(){

        Flux.just(1, 2, 3)
                .doOnComplete(() -> System.out.println("doOnComplete"))
                .subscribe(System.out::println);
//输出为：1 2 3 doOnComplete


        Mono.just("Hello, World!")

                .doOnSuccess(s -> System.out.println("Successfully emitted: " + s))
                .subscribe(result -> System.out.println("Received: " + result));
//输出为：doOnSubscribe: reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber@4b1b1c6a 1 2 3





    }

    static int count=10; //开关总次数
    static int SLEEP=2000; //整体间隔
    static int END=5000; //开启关闭间隔

   static int startErrorCount =1; //开始的重试次数
    static int endErrorCount =1; //结束的重试次数
    public static void main(String[] args) throws InterruptedException {
       // test06();
       // Thread.sleep(5000);
       // complete.dispose();
        //Thread.sleep(1000000000);

      //  sendRequest();
       // Flux.concat(Flux.just("a"),Flux.just("d","e","f"))
         //       .subscribe(System.out::println);
//        Flux.just("a","b","c").zipWith(Flux.just("d","e","f"))
//                .subscribe(System.out::println);
        //sendRequest();
       sendApp();
//       Mono.just("1")
//               .delayElement(Duration.ofSeconds(4))
//        //       .concatWith( Mono.error(new RuntimeException()))
//
//                .doOnError(Throwable::printStackTrace)
//                .subscribe(System.out::println);
//
//        System.out.println("----");
    }

    private static void sendApp() throws InterruptedException{


        String start="http://www.njjinrui.cn:7002/mb-terminal-service/device/register/wechat/charge/99902411100002/true";

        //String start = "http://www.njjinrui.cn:7002/pile-manage-service/login/device/apply/charge";
        HashMap<String,Object> params = new HashMap<>();
        params.put("deviceCode","99902411100002");
        params.put("startMode",2);
        params.put("rateId",1);
       // String end ="http://www.njjinrui.cn:7002/mb-terminal-service/device/register/uncharge/99902411100002";
        List<String> cookieList = new ArrayList<>();
        cookieList.add("LP=eGhQTcf24CMBlly6TE2QiR9ftLNbegT1BegQdHe_geBxnVVSztYWReByhTZm2DU18o0a04a2N-GIlOp6TmD3FIA9ZedAS3BrcVJm6xZCEBCBe-ay6kWfJVXzyZv3FoKPYIdiHVwpuaQOAAdB7JjNzq8rnfvp2BbOG70qEkkDYdeTwmjvbwkNkn2GqMZx8a0YX1QciOBXKfHE9n-MnvMSpS2u9EXEMTMTXE1AR1sh_PgIo6BvP4Ms-ssNQsjj5eFl");
        cookieList.add("Admin-Token=b115eefc-36ea-494d-aecd-4cf2a1e128a5");

        HttpHeaders requestHeaders = new HttpHeaders();
        requestHeaders.put("Cookie", cookieList);
        requestHeaders.set("appId", "a65acfec56174d968934c6fba576e0dd");
        requestHeaders.setContentType(MediaType.APPLICATION_JSON);
       // HttpEntity<HashMap<String,Object>> requestEntity = new HttpEntity<HashMap<String,Object>>(params, requestHeaders);
        HttpEntity<String> requestEntity = new HttpEntity<String>(null, requestHeaders);

        RestTemplate restTemplate = new RestTemplate();
        for (int i = 0; i < 10; i++) {
          //  System.out.println("第"+i+"次请求");
            new Thread(()->start0(restTemplate, start, requestEntity)).start();
          //  Thread.sleep(END);
          //  if(success){
//                end0(restTemplate, end, requestEntity);
            //    end0(restTemplate, end, requestEntity);
           //     Thread.sleep(SLEEP);
         //   }

        }

    }
//    private static void sendRequest() throws InterruptedException {
//        String start="http://localhost:9527/dev-api/device/register/charge/99902411100002/9";
//        String end ="http://localhost:9527/dev-api/device/register/uncharge/99902411100002";
//        List<String> cookieList = new ArrayList<>();
//        cookieList.add("JSESSIONID=df769297-c1f3-4588-ab93-4961d499df88");
//        cookieList.add("Admin-Token=df769297-c1f3-4588-ab93-4961d499df88");
//        cookieList.add("sidebarStatus=1");
//
//        HttpHeaders requestHeaders = new HttpHeaders();
//        requestHeaders.put("Cookie", cookieList);
//        HttpEntity<String> requestEntity = new HttpEntity<String>(null, requestHeaders);
//
//
//        RestTemplate restTemplate = new RestTemplate();
//        for (int i = 0; i < count; i++) {
//            System.out.println("第"+i+"次请求");
//            Boolean success = start0(restTemplate, start, requestEntity);
//            Thread.sleep(END);
//            if(success){
////                end0(restTemplate, end, requestEntity);
//                end0(restTemplate, end, requestEntity);
//                Thread.sleep(SLEEP);
//            }
//
//        }
//    }

    private static Boolean start0(RestTemplate restTemplate,String url,  HttpEntity<String> requestEntity ){

        JSONObject startObj = restTemplate.exchange(url, HttpMethod.PUT, requestEntity, JSONObject.class).getBody();
  //      JSONObject startObj = restTemplate.exchange(url, HttpMethod.POST, requestEntity, JSONObject.class).getBody();
        Boolean s = (Boolean) startObj.get("success");
        if(!s){
            System.out.println("启动失败重试次数"+startErrorCount);
            System.out.println("失败原因"+startObj.get("message"));
        //    startErrorCount++;
            return false;
//            try {
//                Thread.sleep(100);
//               return start0(restTemplate,url,requestEntity);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
        }else{
            return true;
        }
    }

    private static  Boolean end0(RestTemplate restTemplate,String url,  HttpEntity<String> requestEntity ){

        JSONObject startObj = restTemplate.exchange(url, HttpMethod.PUT, requestEntity, JSONObject.class).getBody();
        Boolean s = (Boolean) startObj.get("success");
        if(!s){
            System.out.println("关闭失败重试次数"+endErrorCount);
            System.out.println("关闭失败原因"+startObj.get("message"));
            endErrorCount++;
            try {
                Thread.sleep(100);
                return end0(restTemplate,url,requestEntity);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }else{
            return true;
        }

    }
}
